package cn.tpshion.pulsar.support;

import lombok.Data;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.api.SubscriptionInitialPosition;
import org.apache.pulsar.client.api.SubscriptionType;

import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

@Data
public class PulsarConsumer {

    public static final String PERSISTENT = "persistent";

    private Schema schema = Schema.STRING;
    private String topic;
    private Set<String> topics = new HashSet<>();
    private String consumerName;
    private String namespace;
    private String tenant;
    private String subscriptionName;
    private SubscriptionType subscriptionType = SubscriptionType.Exclusive;
    private int receiverQueueSize = 1000;
    private int ackTimeoutMillis = 1000;
    private SubscriptionInitialPosition subscriptionInitialPosition = SubscriptionInitialPosition.Earliest;

    public String buildTopic(){
        return PERSISTENT + "://" + tenant + "/" + namespace + "/" + topic;
    }

    public String buildTopic(String topic){
        return PERSISTENT + "://" + tenant + "/" + namespace + "/" + topic;
    }

    public List<String> buildTopics(){
        List<String> list = new ArrayList<>();
        for (String s : topics) {
            list.add(buildTopic(s));
        }
        return list;
    }

}
